package com.camemax.operators;

import com.camemax.pojo.AdvertisementViewCount;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
/*
* Flink WindowFunction 实现
* */
public class AdWindowFunction implements WindowFunction<Long , AdvertisementViewCount, String, TimeWindow> {
    @Override
    public void apply(String provinceName, TimeWindow window, Iterable<Long> input, Collector<AdvertisementViewCount> out) throws Exception {
        // 考虑到SimpleDateFormat存在线程安全问题，使用LocalDateTime && DateTimeFormatter换算日期
        String currentDate =
            LocalDateTime.ofEpochSecond(window.getEnd()/1000, 0, ZoneOffset.ofHours(8))
                    .format(DateTimeFormatter.ofPattern("yyyy/MM/dd HH:mm:ss"));
        out.collect(new AdvertisementViewCount(
            provinceName,
            input.iterator().next(),
            currentDate)
        );
    }
}
